package com.zhike.function;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.zhike.utils.DimUtils;
import com.zhike.utils.DruidDSUtils;
import com.zhike.utils.PhoenixUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.SQLException;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/19 14:50
 * @Description: 自定义sink将数据写入phoenix
 *
 */
public class PhoenixSink extends RichSinkFunction<JSONObject> {

    private DruidDataSource dataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        //创建连接池
        dataSource = DruidDSUtils.createDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //1.获取目标表名
        String sinkTable = value.getString("sinkTable");

        //2.获取连接对象
        DruidPooledConnection connection = dataSource.getConnection();

        //3.写入数据
        JSONObject after = value.getJSONObject("after");
        JSONObject before = value.getJSONObject("before");
        if (before!= null){
            DimUtils.delDimInfo(sinkTable,before.getString("vin"));
        }
        if (after != null) {
            PhoenixUtils.insertValues(connection,sinkTable,after);
        }
        connection.close();

    }
}
